package com.feidee.fd.sml.algorithm.component.feature

import org.apache.spark.ml.PipelineStage
import org.apache.spark.ml.feature.ElementwiseProduct
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.DataFrame
/**
  * @author YongChen
  * @date 2019/3/22 3:18
  * @description
  * @reviewer songhaicheng
  */
case class ElementwiseProductorParam(
                                      override val input_pt: String,
                                      override val output_pt: String,
                                      override val hive_table: String,
                                      override val flow_time: String,
                                      override val inputCol: String,
                                      override val outputCol: String,
                                      override val preserveCols: String,
                                      override val modelPath: String ,
                                      // 用来与输入列进行相乘的向量
                                      scalingVec : Array[Double]
                                    ) extends FeatureParam {

  def this() = this(null, null, null, null, "input", "features", null, null, Array.empty[Double])

  override def verify(): Unit = {
    super.verify()
    require(scalingVec.nonEmpty, "param scalingVec can't be null")
  }

  override def toMap: Map[String, Any] = {
    var map = super.toMap
    map += ("scalingVec" -> scalingVec)
    map
  }
}


class ElementwiseProductor  extends AbstractFeatureEncoder[ElementwiseProductorParam]{

  override def setUp(param: ElementwiseProductorParam, data: DataFrame): Array[PipelineStage] = {
  val etp = new ElementwiseProduct()
    .setInputCol(param.inputCol)
    .setOutputCol(param.outputCol)

    if(tool.isNotNull(param.scalingVec)){
      etp.setScalingVec(Vectors.dense(param.scalingVec))
    }

    Array(etp)
  }
}

object ElementwiseProductor {

  def apply(paramStr: String): Unit = {
    new ElementwiseProductor()(paramStr)
  }

  def main(args: Array[String]): Unit = {
    ElementwiseProductor(args(0))
  }

}


